package com.study.flink.demo;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.CsvReader;
import org.apache.flink.api.java.operators.DataSource;

import com.study.flink.entity.does.ZomatoDO;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class ZomatoDemo {
    // private static final String FILE_PATH = "/data/bigdata/zomato.csv";
    private static final String FILE_PATH = "data/demo.csv";

    public void run() {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        CsvReader reader = new CsvReader(FILE_PATH, env);
        reader.ignoreFirstLine();
        DataSource<ZomatoDO> source = reader.pojoType(ZomatoDO.class, "url", "address", "name",
            "onlineOrder", "bookTable", "rate", "votes", "phone", "location", "restType",
            "dishLiked", "cuisines", "approxCost", "reviewsList", "menuItem", "listedInType",
            "listedInCity");

        final AtomicLong index = new AtomicLong(0);

        try {
            List<ZomatoDO> list = new ArrayList<ZomatoDO>();
            // List<ZomatoDO> list = source.filter(new FilterFunction<ZomatoDO>() {
            // @Override
            // public boolean filter(ZomatoDO value) throws Exception {
            // return false;
            // }
            // }).collect();
            long size = source.count();

            env.execute();
            log.info("the data size in stream is: " + size);

            for (ZomatoDO zomato : list) {
                log.info("Zomato: " + zomato.toString());
            }
        }
        catch (Exception e) {
            e.printStackTrace();
        }
        
        log.info("the data size by filter is: " + index.longValue());
    }
}
